// BSD License (http://www.galagosearch.org/license)

package org.galagosearch.tupleflow.execution;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.galagosearch.tupleflow.execution.StageInstanceDescription.PipeInput;
import org.galagosearch.tupleflow.execution.StageInstanceDescription.PipeOutput;

/**
 *
 * @author trevor
 */
public class StageGroupDescription {
    /// A stage object (probably genereated by the JobConstructor parser)
    Stage stage;
    
    public HashMap<String, DataPipeRegion> inputs;
    public HashMap<String, DataPipeRegion> outputs;
    
    /// Count of instances of this stage
    public int instanceCount;

    /// URL of the Master for this job.
    String masterURL;
    
    public static class DataPipeRegion {
        DataPipe pipe;
        int start;
        int end;
        ConnectionPointType direction;
        
        public DataPipeRegion(DataPipe pipe, int start, int end, ConnectionPointType direction) {
            this.pipe = pipe;
            this.start = start;
            this.end = end;
            this.direction = direction;
        }
        
        public int fileCount() {
            int count = 0;
            
            for(int i=start; i<end; i++) {
                String[] filenames = pipe.getOutputFileNames(i);
                count += filenames.length;
            }
            
            return count;
        }
    }
    
    public StageGroupDescription(Stage stage) {
        this(stage, 1, "");
    }
    
    /** Creates a new instance of StageGroupDescription */
    public StageGroupDescription(Stage stage, int instanceCount, String masterURL) {
        this.stage = stage;
        this.inputs = new HashMap<String, DataPipeRegion>();
        this.outputs = new HashMap<String, DataPipeRegion>();
        this.instanceCount = instanceCount;
        this.masterURL = masterURL;
    }
    
    public String getName() {
        return stage.name;
    }

    public String getMasterURL() {
        return masterURL;
    }

    public void setMasterURL(String masterURL) {
        this.masterURL = masterURL;
    }

    public boolean containsInput(String name) {
        return inputs.containsKey(name);
    }
    
    public boolean containsOutput(String name) {
        return outputs.containsKey(name);
    }

    public Stage getStage() {
        return stage;
    }

    public int getInstanceCount() {
        return instanceCount;
    }

    @Override
    public String toString() {
        return stage.toString();
    }


    public List<StageInstanceDescription> getInstances() {
        ArrayList<StageInstanceDescription> instances = new ArrayList();
        
        for(int i=0; i<instanceCount; i++) {
            Map<String, PipeInput> instanceOutputs = new HashMap<String, PipeInput>();
            
            for(String key : outputs.keySet()) {
                DataPipeRegion region = outputs.get(key);
                
                if(region.end - region.start <= 1) {
                    instanceOutputs.put(key, new PipeInput(region.pipe, 0));
                } else {
                    assert region.end - region.start == instanceCount;
                    instanceOutputs.put(key, new PipeInput(region.pipe, region.start + i));
                }
            }
            
            Map<String, PipeOutput> instanceInputs = new HashMap<String, PipeOutput>();

            for(String key : inputs.keySet()) {
                DataPipeRegion region = inputs.get(key);
                
                if(region.end - region.start <= 1) {
                    instanceInputs.put(key, new PipeOutput(region.pipe, 0));
                } else if(instanceCount == 1 && region.end - region.start > 1) {
                    // assignment == "combined"
                    instanceInputs.put(key, new PipeOutput(region.pipe, region.start, region.end));
                } else {
                    assert region.end - region.start == instanceCount;
                    instanceInputs.put(key, new PipeOutput(region.pipe, region.start + i));
                }
            }

            StageInstanceDescription instance =
                    new StageInstanceDescription(
                            stage, i, instanceOutputs, instanceInputs, masterURL);
            instances.add(instance);
        }
        
        return instances;
    }
}
